package io.openmessaging.storage.dledger;

import com.alibaba.fastjson.JSON;
import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode;
import io.openmessaging.storage.dledger.protocol.HeartBeatRequest;
import io.openmessaging.storage.dledger.protocol.HeartBeatResponse;
import io.openmessaging.storage.dledger.protocol.LeadershipTransferRequest;
import io.openmessaging.storage.dledger.protocol.LeadershipTransferResponse;
import io.openmessaging.storage.dledger.protocol.VoteRequest;
import io.openmessaging.storage.dledger.protocol.VoteResponse;
import io.openmessaging.storage.dledger.utils.DLedgerUtils;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//基于Raft协议的Leader选举类
public class DLedgerLeaderElector {

    public static Logger logger = LoggerFactory.getLogger(DLedgerLeaderElector.class);

    //随机数生成器，对应Raft协议中选举超时时间，是一个随机数
    public Random random = new Random();
    //配置参数
    public DLedgerConfig dLedgerConfig;
    //节点状态机
    public final MemberState memberState;
    //RPC服务，实现向集群内的节点发送心跳包、投票的RPC
    public DLedgerRpcService dLedgerRpcService;

    //as a server handler
    //record the last leader state
    //上次收到心跳包的时间戳
    public volatile long lastLeaderHeartBeatTime = -1;
    //上次发送心跳包的时间戳
    public volatile long lastSendHeartBeatTime = -1;
    //上次成功收到心跳包的时间戳
    public volatile long lastSuccHeartBeatTime = -1;
    //一个心跳包的周期，默认为2s
    public int heartBeatTimeIntervalMs = 2000;
    //允许最大的n个心跳周期内未收到心跳包，
    // 状态为Follower的节点只有超过 maxHeartBeatLeak * heartBeatTimeIntervalMs 的时间内未收到主节点的心跳包，
    // 才会重新进入Candidate状态，进行下一轮选举
    public int maxHeartBeatLeak = 3;
    //as a client
    // 发送下一个心跳包的时间戳
    public long nextTimeToRequestVote = -1;
    //是否应该立即发起投票。如果为true，则忽略计时器，该值默认为false
    // 作用是在从节点收到主节点的心跳包，
    // 并且当前状态机的轮次大于主节点轮次（说明集群中Leader的投票轮次小于从节点的轮次）时，立即发起新的投票请求
    public volatile boolean needIncreaseTermImmediately = false;
    //最小的发送投票间隔时间，默认为300ms
    public int minVoteIntervalMs = 300;
    //最大的发送投票间隔时间，默认为1000ms
    public int maxVoteIntervalMs = 1000;

    //注册的节点 状态处理器，通过 addRoleChangeHandler() 方法添加
    public List<RoleChangeHandler> roleChangeHandlers = new ArrayList<>();
    //投票结果 默认是 WAIT_TO_REVOTE
    public VoteResponse.ParseResult lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
    //上一次投票的开销
    public long lastVoteCost = 0L;
    //状态机管理器
    public StateMaintainer stateMaintainer = new StateMaintainer("StateMaintainer", logger);

    //
    private final TakeLeadershipTask takeLeadershipTask = new TakeLeadershipTask();

    public DLedgerLeaderElector(DLedgerConfig dLedgerConfig, MemberState memberState, DLedgerRpcService dLedgerRpcService) {
        this.dLedgerConfig = dLedgerConfig;
        this.memberState = memberState;
        this.dLedgerRpcService = dLedgerRpcService;
        refreshIntervals(dLedgerConfig);
    }

    //选举状态管理器初始化
    public void startup() {
        //1. stateMaintainer是Leader选举内部维护的状态机，
        // 即维护节点状态在Follower、Candidate、Leader之间转换，
        // 需要先调用其 start() 方法启动状态机
        stateMaintainer.start();
        //2. 依次启动注册的角色转换监听器，即内部状态机的状态发生变更后的事件监听器，是Leader选举的功能扩展点
        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            roleChangeHandler.startup();
        }
    }

    public void shutdown() {
        stateMaintainer.shutdown();
        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            roleChangeHandler.shutdown();
        }
    }

    //
    private void refreshIntervals(DLedgerConfig dLedgerConfig) {
        this.heartBeatTimeIntervalMs = dLedgerConfig.getHeartBeatTimeIntervalMs();
        this.maxHeartBeatLeak = dLedgerConfig.getMaxHeartBeatLeak();
        this.minVoteIntervalMs = dLedgerConfig.getMinVoteIntervalMs();
        this.maxVoteIntervalMs = dLedgerConfig.getMaxVoteIntervalMs();
    }

    //从节点 处理接收到的 主节点-心跳包请求
    public CompletableFuture<HeartBeatResponse> handleHeartBeat(HeartBeatRequest request) throws Exception {
        //排外
        if (!memberState.isPeerMember(request.getLeaderId())) {
            logger.warn("[BUG] [HandleHeartBeat] remoteId={} is an unknown member", request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNKNOWN_MEMBER.getCode()));
        }
        //没有自己向自己发心跳包的
        if (memberState.getSelfId().equals(request.getLeaderId())) {
            logger.warn("[BUG] [HandleHeartBeat] selfId={} but remoteId={}", memberState.getSelfId(), request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNEXPECTED_MEMBER.getCode()));
        }

        // 第一步 : 如果发送心跳包的节点（Leader节点）的投票轮次小于从节点的投票轮次，
        // 返回EXPIRED_TERM，告知对方它的投票轮次已经过期，需要重新进入选举
        if (request.getTerm() < memberState.currTerm()) {
            //说明 这个请求 过期了
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
        }
        //在同一个轮次  能交流
        else if (request.getTerm() == memberState.currTerm()) {
            //发来的主节点心跳包中的 leader 节点 跟 当前从节点认定的 leader 节点  相同一样 一家人
            if (request.getLeaderId().equals(memberState.getLeaderId())) {
                //更新心跳时间
                lastLeaderHeartBeatTime = System.currentTimeMillis();
                //返回空的 心跳包响应
                return CompletableFuture.completedFuture(new HeartBeatResponse());
            } else {
                // ! request.getLeaderId().equals(memberState.getLeaderId())
                //1. 轮次相同下   两者LeaderId 就是不相同
                //2. 轮次相同下   当前从节点 的主节点Id 为 null
            }
        }

        //通常情况下第一步将直接返回，本步骤主要用于处理异常情况，需要加锁以确保线程安全

        //执行到这里 有 3种情况 :
        //1. request.getTerm() >  memberState.currTerm()  //轮次直接不同   主节点的轮次 > 当前节点的轮次
        //2. ! request.getLeaderId().equals(memberState.getLeaderId()) //轮次相同下   两者 LeaderId 就是不相同
        //3. ! request.getLeaderId().equals(memberState.getLeaderId()) //轮次相同下   当前从节点 的主节点Id 为 null
        //执行如下加锁逻辑 ↓

        //abnormal case
        //hold the lock to get the latest term and leaderId
        synchronized (memberState) {
            //再次判断  当前轮次是否 大于 主节点轮次
            if (request.getTerm() < memberState.currTerm()) {
                //大于 就说明 主节点 过期了 开下一把吧 需要重新进入选举
                return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
            }
            //两个在同一个轮次  能交流
            else if (request.getTerm() == memberState.currTerm()) {
                //当前节点还没 找到 leader 节点 (上述第3种情况)
                if (memberState.getLeaderId() == null) {
                    //正好 收到了 来自 leader 节点的心跳节点 那就自己变为 follower 节点吧
                    changeRoleToFollower(request.getTerm(), request.getLeaderId());
                    return CompletableFuture.completedFuture(new HeartBeatResponse());
                }
                //一家人
                else if (request.getLeaderId().equals(memberState.getLeaderId())) {
                    //继续更新 心跳时间
                    lastLeaderHeartBeatTime = System.currentTimeMillis();
                    return CompletableFuture.completedFuture(new HeartBeatResponse());
                }
                // !request.getLeaderId().equals(memberState.getLeaderId()  // 轮次相同下 两个 LeaderId 不相等
                else {
                    //轮次相同下   两者 LeaderId 就是不相同 (上述第2种情况)
                    //说明集群中存在另外一个Leader节点，则返回 INCONSISTENT_LEADER，对端节点将进入 Candidate 状态
                    //this should not happen, but if happened
                    logger.error("[{}][BUG] currTerm {} has leader {}, but received leader {}", memberState.getSelfId(), memberState.currTerm(), memberState.getLeaderId(), request.getLeaderId());
                    //不应该发生 但是还是发生了
                    return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.INCONSISTENT_LEADER.getCode()));
                }
            }
            // request.getTerm() > memberState.currTerm() //轮次直接不同   主节点的轮次 > 当前节点的轮次
            //说明 则认为从节点并未准备好，从节点将进入Candidate状态，并立即发起一次投票
            else {
                // To make it simple, for larger term, do not change to follower immediately first change to candidate,
                //为了简单起见，对于更大的任期，不要立即更改为关注者，首先更改为候选人
                // and notify the state-maintainer thread
                //并通知状态维护线程
                //针对更大的周期 选择 把当前节点 转变为 选举者 状态
                changeRoleToCandidate(request.getTerm());
                //主节点的轮次 > 当前节点的轮次
                needIncreaseTermImmediately = true;
                //TODO notify
                return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.TERM_NOT_READY.getCode()));
            }
        }
    }

    //当状态机从Candidate状态变更为Leader节点后会调用该方法，
    //当处于Candidate状态的节点在得到集群内超过半数节点的支持后将进入该状态
    //term : 在这个轮次下 统计的结果 导致要调用这个函数
    public void changeRoleToLeader(long term) {
        synchronized (memberState) {
            //首先更新状态机（MemberState）的角色为Leader，
            // 并设置 leaderId 为当前节点的 ID，然后调用 handleRoleChange() 方法触发角色
            //状态转换事件，从而执行扩展点的逻辑代码
            if (memberState.currTerm() == term) {
                //清除 存活列表   peersLiveTable.clear();
                memberState.changeToLeader(term);
                lastSendHeartBeatTime = -1;
                handleRoleChange(term, MemberState.Role.LEADER);
                logger.info("[{}] [ChangeRoleToLeader] from term: {} and currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            } else {
                logger.warn("[{}] skip to be the leader in term: {}, but currTerm is: {}", memberState.getSelfId(), term, memberState.currTerm());
            }
        }
    }

    //term 已知集群中的最大轮次
    public void changeRoleToCandidate(long term) {
        //锁住
        synchronized (memberState) {
            //传进来的 轮次 > 当前节点的轮次  说明什么? 说明当前节点需要更新了
            if (term >= memberState.currTerm()) {
                //不管当前是什么角色 都转为 candidate 角色
                memberState.changeToCandidate(term);
                handleRoleChange(term, MemberState.Role.CANDIDATE);
                logger.info("[{}] [ChangeRoleToCandidate] from term: {} and currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            } else {
                logger.info("[{}] skip to be candidate in term: {}, but currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            }
        }
    }

    //just for test
    public void testRevote(long term) {
        changeRoleToCandidate(term);
        lastParseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
        nextTimeToRequestVote = -1;
    }

    public void changeRoleToFollower(long term, String leaderId) {
        logger.info("[{}][ChangeRoleToFollower] from term: {} leaderId: {} and currTerm: {}", memberState.getSelfId(), term, leaderId, memberState.currTerm());
        lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
        memberState.changeToFollower(term, leaderId);
        lastLeaderHeartBeatTime = System.currentTimeMillis();
        handleRoleChange(term, MemberState.Role.FOLLOWER);
    }

    // 因为一个节点可能会收到多个节点的“拉票”请求，存在并发问题，
    // 所以需要引入synchronized机制，锁定状态机memberState对象
    public CompletableFuture<VoteResponse> handleVote(VoteRequest request, boolean self) {
        //hold the lock to get the latest term, leaderId, ledgerEndIndex
        synchronized (memberState) {
            //外来的节点 不允许参与投票
            if (!memberState.isPeerMember(request.getLeaderId())) {
                logger.warn("[BUG] [HandleVote] remoteId={} is an unknown member", request.getLeaderId());
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_UNKNOWN_LEADER));
            }
            //不是自己投自己  && LeaderId 是自己 = 那么就发生错误 因为一般都是自己投自己
            if (!self && memberState.getSelfId().equals(request.getLeaderId())) {
                logger.warn("[BUG] [HandleVote] selfId={} but remoteId={}", memberState.getSelfId(), request.getLeaderId());
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_UNEXPECTED_LEADER));
            }
            //拒绝票，原因是自己维护的 ledgerTerm 小于对端维护的 ledgerTerm，
            if (request.getLedgerEndTerm() < memberState.getLedgerEndTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                        voteResult(VoteResponse.RESULT.REJECT_EXPIRED_LEDGER_TERM));
            }
            //拒绝票，原因是对端的 ledgerTeam 与自己维护的 ledgerTeam 相等，轮次相等
            //但自己维护的 LedgerEndIndex 小于对端维护的值， 日志序列小了
            else if (request.getLedgerEndTerm() == memberState.getLedgerEndTerm() &&
                    request.getLedgerEndIndex() < memberState.getLedgerEndIndex()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                        voteResult(VoteResponse.RESULT.REJECT_SMALL_LEDGER_END_INDEX));
            }

            //1. 发起投票节点的投票轮次 小于 当前节点的投票轮次：投拒绝票，也就是说在Raft协议中，term 越大，越有话语权
            if (request.getTerm() < memberState.currTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_EXPIRED_VOTE_TERM));
            }
            //2. 发起投票节点的投票轮次等于当前节点的投票轮次：
            // 说明两者都处在同一个投票轮次中，地位平等，接下来看该节点是否已经投过票
            else if (request.getTerm() == memberState.currTerm()) {
                //当前节点没有投过票
                if (memberState.currVoteFor() == null) {
                    //let it go
                }
                //投出去了 也是投的 Leader 票  那么就相安无事
                else if (memberState.currVoteFor().equals(request.getLeaderId())) {
                    //repeat just let it go
                }
                //不为空 && 也不是投的 Leader 票
                else {
                    //如果该节点已有 Leader 节点
                    //一轮只能投出去 一票
                    if (memberState.getLeaderId() != null) {
                        return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                                voteResult(VoteResponse.RESULT.REJECT_ALREADY_HAS_LEADER));
                    } else {
                        //如果该节点还未有 Leader 节点，但已经投了其他节点的票，则拒绝请求节点，并告知 已投票
                        return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                                voteResult(VoteResponse.RESULT.REJECT_ALREADY_VOTED));
                    }
                }
            }
            // request.getTerm() > memberState.currTerm()
            else {
                // 发起投票节点的 投票轮次 大于 当前节点的投票轮次：
                // 拒绝发起投票节点的投票请求，并告知对方自己还未准备投票，突如其来的骑请求 , 防不胜防 ;
                // 会使用发起投票节点的投票轮次立即进入 Candidate 状态
                // stepped down by larger term
                changeRoleToCandidate(request.getTerm());
                // 发起投票节点的投票轮次 > 当前节点的投票轮次
                needIncreaseTermImmediately = true;
                //only can handleVote when the term is consistent
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                        voteResult(VoteResponse.RESULT.REJECT_TERM_NOT_READY));
            }

            //2. 根据发起投票节点、当前响应节点维护的 ledgerEndTerm 进行投票仲裁
            // 如果发起投票节点的ledgerEndTerm小于当前响应节点的
            // ledgerEndTerm则拒绝，原因是发起投票节点的日志复制进度比当前节点低，
            // 这种情况是不能成为主节点的，否则会造成数据丢失
            if (request.getTerm() < memberState.getLedgerEndTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.getLedgerEndTerm()).
                        voteResult(VoteResponse.RESULT.REJECT_TERM_SMALL_THAN_LEDGER));
            }

            //如果发起投票节点的ledgerEndTerm与当前响应节点维护的
            //ledgerEndTerm相等，但是ledgerEndIndex比当前节点小，则拒绝，原因与上一条相同
            if (!self && isTakingLeadership() && request.getLedgerEndTerm() == memberState.getLedgerEndTerm() &&
                    memberState.getLedgerEndIndex() >= request.getLedgerEndIndex()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).
                        voteResult(VoteResponse.RESULT.REJECT_TAKING_LEADERSHIP));
            }

            //3, 经过层层筛选，将宝贵的赞成票投给发起投票的节点，完成整个投票过程
            memberState.setCurrVoteFor(request.getLeaderId());
            return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.ACCEPT));
        }
    }

    //经过几轮投票，其中一个节点会被推举出来成为Leader节点。
    //Leader节点为了维持其领导地位，会定时向从节点发送心跳包
    //并处理 自己刚发出的心跳包的响应
    private void sendHeartbeats(long term, String leaderId) throws Exception {
        //集群内节点的总个数
        final AtomicInteger allNum = new AtomicInteger(1);
        //集群内节点的发送成功的个数
        final AtomicInteger succNum = new AtomicInteger(1);
        //收到对端没有准备好反馈的节点个数
        final AtomicInteger notReadyNum = new AtomicInteger(0);
        //当前集群中各个节点维护的最大的投票轮次
        final AtomicLong maxTerm = new AtomicLong(-1);
        //是否存在Leader节点不一致
        final AtomicBoolean inconsistLeader = new AtomicBoolean(false);
        //用于等待异步请求结果
        final CountDownLatch beatLatch = new CountDownLatch(1);
        //本次心跳包开始发送的时间戳
        long startHeartbeatTimeMs = System.currentTimeMillis();
        //遍历发送心跳
        for (String id : memberState.getPeerMap().keySet()) {
            if (memberState.getSelfId().equals(id)) {
                continue;
            }
            //第一步：遍历集群中所有的节点，构建心跳数据包并异步向集群内的从节点发送心跳包，
            // 心跳包中主要包含Raft复制组名、当前节点ID、远程节点ID、当前集群中的leaderId、当前节点维护的投票轮次
            HeartBeatRequest heartBeatRequest = new HeartBeatRequest();
            heartBeatRequest.setGroup(memberState.getGroup());
            heartBeatRequest.setLocalId(memberState.getSelfId());
            heartBeatRequest.setRemoteId(id);
            heartBeatRequest.setLeaderId(leaderId);
            heartBeatRequest.setTerm(term);
            //底层网络发送实现 心跳包
            CompletableFuture<HeartBeatResponse> future = dLedgerRpcService.heartBeat(heartBeatRequest);
            //当收到心跳包的响应时
            future.whenComplete((HeartBeatResponse x, Throwable ex) -> {
                try {
                    //异常情况
                    if (ex != null) {
                        memberState.getPeersLiveTable().put(id, Boolean.FALSE);
                        throw ex;
                    }
                    //第二步：当收到一个节点的响应结果后触发回调函数，统计响应结果，先介绍一下对端节点的返回结果
                    switch (DLedgerResponseCode.valueOf(x.getCode())) {
                        case SUCCESS://心跳包成功响应
                            succNum.incrementAndGet();
                            break;
                        case EXPIRED_TERM://当前 Leader 节点的投票轮次 小于 从节点的投票轮次
                            maxTerm.set(x.getTerm());
                            break;
                        case INCONSISTENT_LEADER://从节点已经有了新的主节点
                            //存在 leader 不一致
                            inconsistLeader.compareAndSet(false, true);
                            break;
                        case TERM_NOT_READY://从节点未准备好
                            notReadyNum.incrementAndGet();
                            break;
                        default:
                            break;
                    }

                    //保存在线节点的状态
                    if (x.getCode() == DLedgerResponseCode.NETWORK_ERROR.getCode())
                        memberState.getPeersLiveTable().put(id, Boolean.FALSE);
                    else memberState.getPeersLiveTable().put(id, Boolean.TRUE);

                    //如果收到SUCCESS的从节点数量超过集群节点的半数，或者收到集
                    //群内所有节点的响应结果后调用CountDownLatch的countDown()方法从而唤醒了主线程，则继续执行后续流程
                    if (memberState.isQuorum(succNum.get()) || memberState.isQuorum(succNum.get() + notReadyNum.get())) {
                        //唤醒主线程
                        beatLatch.countDown();
                    }
                } catch (Throwable t) {
                    logger.error("heartbeat response failed", t);
                } finally {
                    allNum.incrementAndGet();
                    if (allNum.get() == memberState.peerSize()) {
                        beatLatch.countDown();
                    }
                }
            });
        }//for over

        // 阻塞 2s
        beatLatch.await(heartBeatTimeIntervalMs, TimeUnit.MILLISECONDS);
        //第三步：对响应结果进行仲裁
        //① 如果当前Leader节点收到超过集群半数节点的认可（SUCCESS），表示集群状态正常，则正常按照心跳包间隔发送心跳包
        if (memberState.isQuorum(succNum.get())) {
            lastSuccHeartBeatTime = System.currentTimeMillis();
        } else {
            //单纯 成功的个数 不足以 半数 那么就用 未准备的个数来
            logger.info("[{}] Parse heartbeat responses in cost={} term={} allNum={} succNum={} notReadyNum={} inconsistLeader={} maxTerm={} peerSize={} lastSuccHeartBeatTime={}", memberState.getSelfId(), DLedgerUtils.elapsed(startHeartbeatTimeMs), term, allNum.get(), succNum.get(), notReadyNum.get(), inconsistLeader.get(), maxTerm.get(), memberState.peerSize(), new Timestamp(lastSuccHeartBeatTime));
            //② 如果当前Leader节点收到 SUCCESS的响应数 + 未准备投票的节点数超过集群节点的半数，则立即发送心跳包
            if (memberState.isQuorum(succNum.get() + notReadyNum.get())) {
                //
                lastSendHeartBeatTime = -1;
            }
            //③ A+B 没有超过半数 &&  集群中最大轮次 > 当前 Leader 节点轮次
            // 如果从节点的投票轮次比主节点大，则使用从节点的投票轮次
            else if (maxTerm.get() > term) {
                //该移位了 Leader 转为 candidate 角色
                changeRoleToCandidate(maxTerm.get());
            }
            //④ A+B 没有超过半数 &&  集群中最大轮次 <= 当前 Leader 节点轮次
            // 判断是否出现了 另外一个 Leader  , 俗称 网络分区
            else if (inconsistLeader.get()) {
                //当前节点改变状态移位
                changeRoleToCandidate(term);
            }
            //⑤ A+B 没有超过半数 &&  集群中最大轮次 <= 当前节点轮次 &&  没有第二个 Leader  \
            // 判断 收到的心跳包的间隔 > 最大间隔了
            else if (DLedgerUtils.elapsed(lastSuccHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs) {
                //移位
                changeRoleToCandidate(term);
            }
        }
    }

    //经过 maintainAsCandidate() 投票选举被其他节点选举为Leader后，
    //在该状态下会执行 maintainAsLeader() 方法，此时 , 其他节点不知情 , 状态还是 Candidate，
    // 并在自身计时器过期后，又会尝试发起选举。
    // 那么接下来重点分析成为Leader节点后，该节点会做些什么 ?
    private void maintainAsLeader() throws Exception {
        //距离上一次的心跳时间间隔 > 默认的 2s
        if (DLedgerUtils.elapsed(lastSendHeartBeatTime) > heartBeatTimeIntervalMs) {
            //当前轮次
            long term;
            //领导ID
            String leaderId;
            //加锁 只有一个 memberState 实例
            synchronized (memberState) {
                //当前节点不是 leader 了 完犊子了
                if (!memberState.isLeader()) {
                    //stop sending
                    return;
                }
                //获得当前轮次
                term = memberState.currTerm();
                //获得当前领导Id
                leaderId = memberState.getLeaderId();
                //获得当前心跳时间错
                lastSendHeartBeatTime = System.currentTimeMillis();
            }
            //发送心跳包
            sendHeartbeats(term, leaderId);
        }
    }

    //我们先来看在Follower状态下，节点会做些什么 ?
    private void maintainAsFollower() {
        //从这里也不得不佩服 RocketMQ 在性能方面如此追求极致，
        // 即在不加锁的情况下判断是否超过了2个心跳包周期，减少加锁次数，提高性能
        if (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > 2 * heartBeatTimeIntervalMs) {
            //申请加锁
            synchronized (memberState) {
                //如果当前是 Follower && 如果上一次 收到的 心跳包时间间隔 > 3 * 2s
                //如果节点在maxHeartBeatLeak个心跳包（默认为3个）周期内未收到心跳包，则将状态变更为 Candidate
                if (memberState.isFollower() && (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs)) {
                    logger.info("[{}][HeartBeatTimeOut] lastLeaderHeartBeatTime: {} heartBeatTimeIntervalMs: {} lastLeader={}", memberState.getSelfId(), new Timestamp(lastLeaderHeartBeatTime), heartBeatTimeIntervalMs, memberState.getLeaderId());
                    //改变角色
                    changeRoleToCandidate(memberState.currTerm());
                }
            }
        }
    }

    //节点的状态为Candidate时会向集群内的其他节点发起投票请求（个人认为理解为拉票更好），
    // 向对方询问是否愿意选举“我”为 Leader，对端节点会根据自己的情况对其投赞成票或拒绝票，
    // 如果投拒绝票，还会给出拒绝的原因
    //1）long term：发起投票节点当前维护的投票轮次
    //2）long ledgerEndTerm：发起投票节点当前维护的最大投票轮次
    //3）long ledgerEndIndex：发起投票节点维护的最大日志条目索引
    private List<CompletableFuture<VoteResponse>> voteForQuorumResponses(long term, long ledgerEndTerm, long ledgerEndIndex) throws Exception {
        //保存每次投票结果的响应
        List<CompletableFuture<VoteResponse>> responses = new ArrayList<>();
        //向所有节点发送请求
        for (String id : memberState.getPeerMap().keySet()) {
            //将参数都包装起来
            VoteRequest voteRequest = new VoteRequest();
            voteRequest.setGroup(memberState.getGroup());
            voteRequest.setLedgerEndIndex(ledgerEndIndex);
            voteRequest.setLedgerEndTerm(ledgerEndTerm);
            //每个投票请求 都把 LeaderId 设置成 自己的 Id
            voteRequest.setLeaderId(memberState.getSelfId());
            voteRequest.setTerm(term);
            voteRequest.setRemoteId(id);
            CompletableFuture<VoteResponse> voteResponse;
            //判断是否是自己
            if (memberState.getSelfId().equals(id)) {
                //自己向自己投上一票 (处理投票请求器)
                voteResponse = handleVote(voteRequest, true);
            } else {
                //async 向其他节点发送
                voteResponse = dLedgerRpcService.vote(voteRequest);
            }
            //保存每一次的投票 结果
            responses.add(voteResponse);
        }//for over
        return responses;
    }

    private boolean isTakingLeadership() {
        //
        if (dLedgerConfig.getPreferredLeaderIds() != null && memberState.getTermToTakeLeadership() == memberState.currTerm()) {
            List<String> preferredLeaderIds = Arrays.asList(dLedgerConfig.getPreferredLeaderIds().split(";"));
            return preferredLeaderIds.contains(memberState.getSelfId());
        }
        return false;
    }

    private long getNextTimeToRequestVote() {
        if (isTakingLeadership()) {
            return System.currentTimeMillis() + dLedgerConfig.getMinTakeLeadershipVoteIntervalMs() + random.nextInt(dLedgerConfig.getMaxTakeLeadershipVoteIntervalMs() - dLedgerConfig.getMinTakeLeadershipVoteIntervalMs());
        }
        return System.currentTimeMillis() + minVoteIntervalMs + random.nextInt(maxVoteIntervalMs - minVoteIntervalMs);
    }

    //Candidate 下 该干的事情
    private void maintainAsCandidate() throws Exception {
        //for candidate
        //nextTimeToRequestVote :
        //1. 下一次可发起投票的时间，如果当前时间小于该值 && 不需要立即发起投票
        // 说明 计时器 未过期，此时无须发起投票
        if (System.currentTimeMillis() < nextTimeToRequestVote && !needIncreaseTermImmediately) {
            return;
        }
        // 执行到这里说明 需要发起投票
        //当前节点的 投票轮次 很重要 ！！！
        long term;
        // 当前最大的投票轮次
        long ledgerEndTerm;
        // 当前日志的最大序列，即下一条日志的开始 index
        long ledgerEndIndex;
        //只有一个 状态机 锁
        synchronized (memberState) {
            //不是这个状态了 直接划走
            if (!memberState.isCandidate()) {
                return;
            }

            //2. 投票轮次的初始化机制：
            // 如果上一次的投票结果为 WAIT_TO_VOTE_NEXT（等待下一轮投票）
            // 或 应该立即发起投票，则通过状态机获取新一轮投票的序号，默认在当前轮次递增1，
            // 并将 lastParseResult 更新为 WAIT_TO_REVOTE（等待投票）
            if (lastParseResult == VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT || needIncreaseTermImmediately) {
                //保留当前 轮次 作为 上一次 的 轮次
                long prevTerm = memberState.currTerm();
                //只要发起投票  就得 更新轮次 || 更新  获取 下一次 轮次
                term = memberState.nextTerm();
                //等待重新投票中
                lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
                logger.info("{}_[INCREASE_TERM] from {} to {}", memberState.getSelfId(), prevTerm, term);
            }
            // WAIT_TO_REVOTE  REVOTE_IMMEDIATELY PASSED
            else {
                //如果上一次的投票结果不是WAIT_TO_VOTE_NEXT，则投票轮次依然为状态机内部维护的投票轮次
                term = memberState.currTerm();
            }
            //当前日志的最大序列
            ledgerEndIndex = memberState.getLedgerEndIndex();
            // 当前最大的投票轮次
            ledgerEndTerm = memberState.getLedgerEndTerm();

        }//synchronized over

        //3. 如果needIncreaseTermImmediately为true，则重置该标记位为false，
        // 并重新设置下一次投票超时时间，其实现逻辑为 当前时间戳+上次投票的开销+最小投票间隔之间的随机值，
        // 这里是Raft协议的一个关键点，即每个节点的投票超时时间引入了随机值
        if (needIncreaseTermImmediately) {
            //下一次 投票时间 引入 随机值
            nextTimeToRequestVote = getNextTimeToRequestVote();
            //重置
            needIncreaseTermImmediately = false;
            return;
        }

        //当前投票时间戳
        long startVoteTimeMs = System.currentTimeMillis();
        //4. 向集群内的其他节点发起投票请求，并等待各个节点的响应结果
        final List<CompletableFuture<VoteResponse>> quorumVoteResponses = voteForQuorumResponses(term, ledgerEndTerm, ledgerEndIndex);

        //已知的最大投票轮次
        final AtomicLong knownMaxTermInGroup = new AtomicLong(term);
        //所有投票数
        final AtomicInteger allNum = new AtomicInteger(0);
        //有效投票数
        final AtomicInteger validNum = new AtomicInteger(0);
        //赞成票数量
        final AtomicInteger acceptedNum = new AtomicInteger(0);
        //未准备投票的节点数量，如果对端节点的投票轮次 小于 发起投票的轮次，
        // 则认为对端未准备好，对端节点使用本轮次进入Candidate状态
        final AtomicInteger notReadyTermNum = new AtomicInteger(0);
        // 发起投票的节点的 ledgerEndTerm 小于对端节点的个数 || 大于当前轮次的节点个数
        final AtomicInteger biggerLedgerNum = new AtomicInteger(0);
        //是否已经存在Leader
        final AtomicBoolean alreadyHasLeader = new AtomicBoolean(false);
        //当前Jv m中投票同步器
        CountDownLatch voteLatch = new CountDownLatch(1);
        //遍历 future.whenComplete() 投票结果 事件
        for (CompletableFuture<VoteResponse> future : quorumVoteResponses) {
            //具体的业务逻辑，即当请求得到对端的响应结果后的回调方法
            future.whenComplete((VoteResponse x, Throwable ex) -> {
                try {
                    if (ex != null) {
                        throw ex;
                    }
                    logger.info("[{}][GetVoteResponse] {}", memberState.getSelfId(), JSON.toJSONString(x));
                    //更新有效投票数
                    if (x.getVoteResult() != VoteResponse.RESULT.UNKNOWN) {
                        validNum.incrementAndGet();
                    }
                    //6. 统计投票结果，后续会根据投票结果决定是否可以成为 Leader，从而决定当前节点的状态
                    //注意: 以防 whenComplete() 并发执行
                    synchronized (knownMaxTermInGroup) {
                        switch (x.getVoteResult()) {
                            case ACCEPT:
                                //赞成票（acceptedNum）加1，只有得到的赞成票超过集群节点数量的一半才能成为 Leader
                                acceptedNum.incrementAndGet();
                                break;
                            case REJECT_ALREADY_VOTED://拒绝票，原因是已经投给了其他节点
                            case REJECT_TAKING_LEADERSHIP:// 拒绝票 , 拒绝被领导
                                break;
                            case REJECT_ALREADY_HAS_LEADER://拒绝票，原因是集群中已经存 在 Leader 节点了
                                //alreadyHasLeader设置为true，无须再判断其他投票结果了，结束本轮投票
                                alreadyHasLeader.compareAndSet(false, true);
                                break;
                            case REJECT_TERM_SMALL_THAN_LEDGER://拒绝票，原因是自己当前维护的term 小于 远端维护的 ledgerEndTerm , 自己吃了闭门羹
                                //如果对端的team大于自己的team，需要记录对端最大的投票轮次，以便更新自己的投票轮次
                            case REJECT_EXPIRED_VOTE_TERM://拒绝票，原因是自己维护的投票 轮次小于远端维护的投票轮次
                                // 并且更新自己维护的投票轮次
                                if (x.getTerm() > knownMaxTermInGroup.get()) {
                                    //跟随 已知的最大 投票轮次
                                    knownMaxTermInGroup.set(x.getTerm());
                                }
                                break;
                            case REJECT_EXPIRED_LEDGER_TERM://拒绝票，原因是自己维护的 ledgerTerm 小于对端维护的 ledgerTerm，
                                // 此种情况下需要增加计数器 biggerLedgerNum 的值
                            case REJECT_SMALL_LEDGER_END_INDEX: //拒绝票，原因是对端的 ledgerTeam与自己维护的ledgerTeam相等，轮次相等
                                // 但自己维护的 LedgerEndIndex 小于对端维护的值， 日志序列小了
                                // 这种情况下需要增加 biggerLedgerNum 计数器的值
                                biggerLedgerNum.incrementAndGet();
                                break;
                            case REJECT_TERM_NOT_READY://拒绝票，原因是对端的投票轮次 小于 自己的投票轮次，
                                // 即对端还未准备好投票。此时对端节点使用自己的投票轮次进入Candidate状态
                                notReadyTermNum.incrementAndGet();
                                break;
                            default:
                                break;
                        }
                    }//synchronized over

                    //这个不会 重复 执行吗? 答 :  必然会重复执行
                    if (alreadyHasLeader.get() || memberState.isQuorum(acceptedNum.get()) ||
                            memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
                        //唤醒 外部 wait() 的线程 也就是 第5步 处的线程 710行代码处
                        voteLatch.countDown();
                    }
                } catch (Throwable t) {
                    logger.error("vote response failed", t);
                } finally {
                    //  每一次 遍历一个节点 +1
                    allNum.incrementAndGet();
                    //即时上述代码中没有执行 voteLatch.countDown(); 最后保底也得执行一次 唤醒操作
                    if (allNum.get() == memberState.peerSize()) {
                        //唤醒 外部 wait() 的线程 也就是 第5步 处的线程
                        voteLatch.countDown();
                    }
                }
            });
        }//for over

        try {
            // 阻塞 此线程   2000 < x < 3000  /ms
            //5. 在第四步异步向集群内的各个节点发送投票请求，接下来需要同步等待所有的响应结果。
            // 这里展示了一种非常优雅的编程技巧，即在收到对端的响应结果后 先触发 Future 的whenComplete()方法 ,
            // 然后判断是否来触发 CountDownLatch.countDown() 方法 , 最终还有 必定触发 countDown() 保底策略 .
            // 在业务处理过程中，如果条件满足则调用CountDownLatch的countDown方法，唤醒await()方法，
            // 使之接受全部响应结果后执行后续逻辑
            voteLatch.await(2000 + random.nextInt(maxVoteIntervalMs), TimeUnit.MILLISECONDS);
        } catch (Throwable ignore) {
            ignore.printStackTrace();
        }

        //投票所花费的时间
        lastVoteCost = DLedgerUtils.elapsed(startVoteTimeMs);
        //第七步：根据投票结果进行仲裁，从而驱动状态机
        VoteResponse.ParseResult parseResult;
        //① 如果对端的投票轮次 大于 当前节点维护的投票轮次，则先重置投票计时器，
        // 然后在定时器到期后使用对端的投票轮次重新进入 Candidate 状态
        if (knownMaxTermInGroup.get() > term) {
            parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
            nextTimeToRequestVote = getNextTimeToRequestVote();
            //更改角色 并执行相关 监听器
            changeRoleToCandidate(knownMaxTermInGroup.get());
        } else if (alreadyHasLeader.get()) {//②存在主节点了 , 那么 当前节点就需要等待 来自主节点的心跳包
            // 如果集群内已经存在Leader节点，
            // 当前节点将继续保持Candidate状态，重置计时器，但这个计时器还需要增加
            //heartBeatTimeIntervalMs * maxHeartBeatLeak，
            // 其中 heartBeatTimeIntervalMs为一次心跳间隔时间，maxHeartBeatLeak为允许丢失的最大心跳包。
            // 增加这个时间是因为集群内既然已经存在 Leader节点了，那么主节点就会在一个心跳周期内发送心跳包，
            // 从节点在收到心跳包后会重置定时器，即阻止 Follower 节点进入 Candidate 状态。
            // 这样做的目的是在指定时间内收到Leader节点的心跳包，从而驱动当前节点的状态由 Candidate 向 Follower 转换
            //因为一旦进入了 Follower 角色 中  就会开始 周期时间 向 Candidate 角色转变
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            //下一次投票时间 + Leader 心跳 最大间隔时长
            nextTimeToRequestVote = getNextTimeToRequestVote() + heartBeatTimeIntervalMs * maxHeartBeatLeak;
        } else if (!memberState.isQuorum(validNum.get())) {
            //③ 如果收到的有效票数未超过半数，则重置计时器并等待重新投票，
            // 注意当前状态为 WAIT_TO_REVOTE，该状态下的特征是下次投票时不增加投票轮次
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            nextTimeToRequestVote = getNextTimeToRequestVote();
        } else if (!memberState.isQuorum(validNum.get() - biggerLedgerNum.get())) {//
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            //为什么还加一个 最大 发送间隔 ???
            nextTimeToRequestVote = getNextTimeToRequestVote() + maxVoteIntervalMs;
        } else if (memberState.isQuorum(acceptedNum.get())) {//投票的数量到了
            //投票通过
            parseResult = VoteResponse.ParseResult.PASSED;
        } else if (memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
            //执行到这里说明 投票的不够 + 未准备的 够了半数 = 那就赶紧执行下一个投票
            parseResult = VoteResponse.ParseResult.REVOTE_IMMEDIATELY;
        } else {
            //其他情况 :
            //如果得到的赞成票加上对端维护的ledgerEndIndex超过半数，则重置计时器，继续本轮选举
            parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
            nextTimeToRequestVote = getNextTimeToRequestVote();
        }

        //保存此次投票结果
        lastParseResult = parseResult;
        logger.info("[{}] [PARSE_VOTE_RESULT] cost={} term={} memberNum={} allNum={} acceptedNum={} notReadyTermNum={} biggerLedgerNum={} alreadyHasLeader={} maxTerm={} result={}", memberState.getSelfId(), lastVoteCost, term, memberState.peerSize(), allNum, acceptedNum, notReadyTermNum, biggerLedgerNum, alreadyHasLeader, knownMaxTermInGroup.get(), parseResult);
        //④如果得到的赞同票超过半数，则成为Leader节点，调用changeRoleToLeader() 方法驱动状态机向Leader状态转换
        if (parseResult == VoteResponse.ParseResult.PASSED) {
            logger.info("[{}] [VOTE_RESULT] has been elected to be the leader in term {}", memberState.getSelfId(), term);
            //改变角色
            changeRoleToLeader(term);
        }
    }

    /**
     * The core method of maintainer. Run the specified logic according to the current role:
     * candidate => propose a vote.
     * 候选者，该状态下的节点会发起投票，尝试选择自己为主节点，选举成功后，不会存在该状态下的节点
     * leader => send heartbeats to followers, and step down to candidate when quorum followers do not respond.
     * 领导者、主节点，该状态下需要定时向从节点发送心跳包，用于传播数据、确保其领导地位
     * follower => accept heartbeats, and change to candidate when no heartbeat from leader.
     * 从节点，该状态下会开启定时器，尝试进入Candidate状态，以便发起投票选举，一旦收到主节点的心跳包，则重置定时器
     * 注意 : 该方法会不停的被调用
     *
     * @throws Exception
     */
    private void maintainState() throws Exception {
        if (memberState.isLeader()) {
            //主节点 需要定时向从节点发送心跳包，用于传播数据、确保其领导地位
            maintainAsLeader();
        } else if (memberState.isFollower()) {
            //从节点 会开启定时器，尝试进入Candidate状态，以便发起投票选举，一旦收到主节点的心跳包，则重置定时器
            maintainAsFollower();
        } else {
            //选举节点 节点会发起投票，尝试选择自己为主节点，选举成功后，不会存在该状态下的节点
            maintainAsCandidate();
        }
    }

    private void handleRoleChange(long term, MemberState.Role role) {
        try {
            takeLeadershipTask.check(term, role);
        } catch (Throwable t) {
            logger.error("takeLeadershipTask.check failed. ter={}, role={}", term, role, t);
        }

        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            try {
                roleChangeHandler.handle(term, role);
            } catch (Throwable t) {
                logger.warn("Handle role change failed term={} role={} handler={}", term, role, roleChangeHandler.getClass(), t);
            }
        }
    }

    public void addRoleChangeHandler(RoleChangeHandler roleChangeHandler) {
        if (!roleChangeHandlers.contains(roleChangeHandler)) {
            roleChangeHandlers.add(roleChangeHandler);
        }
    }

    public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(LeadershipTransferRequest request) throws Exception {
        logger.info("handleLeadershipTransfer: {}", request);
        synchronized (memberState) {
            if (memberState.currTerm() != request.getTerm()) {
                logger.warn("[BUG] [HandleLeaderTransfer] currTerm={} != request.term={}", memberState.currTerm(), request.getTerm());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.INCONSISTENT_TERM.getCode()));
            }

            if (!memberState.isLeader()) {
                logger.warn("[BUG] [HandleLeaderTransfer] selfId={} is not leader", request.getLeaderId());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.NOT_LEADER.getCode()));
            }

            if (memberState.getTransferee() != null) {
                logger.warn("[BUG] [HandleLeaderTransfer] transferee={} is already set", memberState.getTransferee());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.LEADER_TRANSFERRING.getCode()));
            }

            memberState.setTransferee(request.getTransfereeId());
        }
        LeadershipTransferRequest takeLeadershipRequest = new LeadershipTransferRequest();
        takeLeadershipRequest.setGroup(memberState.getGroup());
        takeLeadershipRequest.setLeaderId(memberState.getLeaderId());
        takeLeadershipRequest.setLocalId(memberState.getSelfId());
        takeLeadershipRequest.setRemoteId(request.getTransfereeId());
        takeLeadershipRequest.setTerm(request.getTerm());
        takeLeadershipRequest.setTakeLeadershipLedgerIndex(memberState.getLedgerEndIndex());
        takeLeadershipRequest.setTransferId(memberState.getSelfId());
        takeLeadershipRequest.setTransfereeId(request.getTransfereeId());
        if (memberState.currTerm() != request.getTerm()) {
            logger.warn("[HandleLeaderTransfer] term changed, cur={} , request={}", memberState.currTerm(), request.getTerm());
            return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
        }

        return dLedgerRpcService.leadershipTransfer(takeLeadershipRequest).thenApply(response -> {
            synchronized (memberState) {
                if (response.getCode() != DLedgerResponseCode.SUCCESS.getCode() || (memberState.currTerm() == request.getTerm() && memberState.getTransferee() != null)) {
                    logger.warn("leadershipTransfer failed, set transferee to null");
                    memberState.setTransferee(null);
                }
            }
            return response;
        });
    }

    public CompletableFuture<LeadershipTransferResponse> handleTakeLeadership(LeadershipTransferRequest request) throws Exception {
        logger.debug("handleTakeLeadership.request={}", request);
        synchronized (memberState) {
            //轮次不一样  不能交流
            if (memberState.currTerm() != request.getTerm()) {
                logger.warn("[BUG] [handleTakeLeadership] currTerm={} != request.term={}", memberState.currTerm(), request.getTerm());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.INCONSISTENT_TERM.getCode()));
            }
            //memberState.currTerm() == request.getTerm()
            long targetTerm = request.getTerm() + 1;
            memberState.setTermToTakeLeadership(targetTerm);
            CompletableFuture<LeadershipTransferResponse> response = new CompletableFuture<>();
            takeLeadershipTask.update(request, response);
            //
            changeRoleToCandidate(targetTerm);
            //
            needIncreaseTermImmediately = true;
            return response;
        }
    }

    private class TakeLeadershipTask {

        private LeadershipTransferRequest request;
        private CompletableFuture<LeadershipTransferResponse> responseFuture;

        public synchronized void update(LeadershipTransferRequest request, CompletableFuture<LeadershipTransferResponse> responseFuture) {
            this.request = request;
            this.responseFuture = responseFuture;
        }

        public synchronized void check(long term, MemberState.Role role) {
            logger.trace("TakeLeadershipTask called, term={}, role={}", term, role);
            if (memberState.getTermToTakeLeadership() == -1 || responseFuture == null) {
                return;
            }
            LeadershipTransferResponse response = null;
            if (term > memberState.getTermToTakeLeadership()) {
                response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.EXPIRED_TERM.getCode());
            } else if (term == memberState.getTermToTakeLeadership()) {
                switch (role) {
                    case LEADER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.SUCCESS.getCode());
                        break;
                    case FOLLOWER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.TAKE_LEADERSHIP_FAILED.getCode());
                        break;
                    default:
                        return;
                }
            } else {
                switch (role) {
                    /*
                     * The node may receive heartbeat before term increase as a candidate,
                     * then it will be follower and term < TermToTakeLeadership
                     */
                    case FOLLOWER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.TAKE_LEADERSHIP_FAILED.getCode());
                        break;
                    default:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.INTERNAL_ERROR.getCode());
                }
            }

            responseFuture.complete(response);
            logger.info("TakeLeadershipTask finished. request={}, response={}, term={}, role={}", request, response, term, role);
            memberState.setTermToTakeLeadership(-1);
            responseFuture = null;
            request = null;
        }
    }

    public interface RoleChangeHandler {
        void handle(long term, MemberState.Role role);

        void startup();

        void shutdown();
    }

    //StateMaintainer的父类为ShutdownAbleThread，继承自Thread，
    //故调用其start()方法最终会调用run()方法
    public class StateMaintainer extends ShutdownAbleThread {

        public StateMaintainer(String name, Logger logger) {
            super(name, logger);
        }

        /**
         * StateMaintainer 状态机的实现要点就是“无限死循环”调用doWork()方法，直到该状态机被关闭。
         * doWork() 方法在 ShutdownAbleThread 被声明为抽象方法，具体由各个子类实现，
         */
        @Override
        public void doWork() {
            try {
                //如果当前节点参与 Leader 选举，则调用 maintainState() 方法驱动状态机，并且每一次驱动状态机后休息 10ms
                if (DLedgerLeaderElector.this.dLedgerConfig.isEnableLeaderElector()) {
                    //由于是上层不断循环的调用 此方法 那么 就说明 在不断的重置 各种时间 间隔
                    DLedgerLeaderElector.this.refreshIntervals(dLedgerConfig);
                    // 状态机的驱动实现思路 : 就是根据状态机当前状态对应的方法，在该状态下，检测状态机是否满足状态变更的条件，如果满足则变更状态
                    DLedgerLeaderElector.this.maintainState();
                }
                sleep(10);
            } catch (Throwable t) {
                DLedgerLeaderElector.logger.error("Error in heartbeat", t);
            }
        }

    }
}
